package com.gy.spark.core;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;

public class SparkWc {

    public static void main(String[] args) {
        System.setProperty("HADOOP_USER_NAME", "spadger");
        System.setProperty("user.name", "spadger");

        SparkConf sparkConf = new SparkConf()
                .setAppName(SparkWc.class.getSimpleName())
                .setMaster("spark://centos101:7077,centos102:7077")
                .setJars(new String[]{"/Users/mac/Downloads/workspace/kkb-bigdata/target/kkb-bigdata-1.0-SNAPSHOT.jar"});

        sparkConf.set("fs.defaultFS", "hdfs://centos102:50070");

        sparkConf.set("spark.driver.host", "192.168.100.1");
        sparkConf.set("spark.submit.deployMode", "client");
        sparkConf.set("spark.executor.memory", "1024M");
        sparkConf.set("spark.executor.cores", "1");

        JavaSparkContext sc = new JavaSparkContext(sparkConf);


        JavaRDD<String> textRdd = sc.textFile("file:///Users/mac/Downloads/workspace/kkb-bigdata/src/main/resources/jdbc.properties", 1);

        JavaRDD<String> wordsRdd = textRdd.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String s) throws Exception {
                return Arrays.asList(s.split(" ")).iterator();
            }
        });

        JavaPairRDD<String, Integer> wordRdd = wordsRdd.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                return new Tuple2<String, Integer>(s, 1);
            }
        });

        JavaPairRDD<String, Integer> reduceRdd = wordRdd.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer integer, Integer integer2) throws Exception {
                return integer + integer2;
            }
        });

        reduceRdd.foreach(new VoidFunction<Tuple2<String, Integer>>() {
            @Override
            public void call(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                System.err.println(stringIntegerTuple2);
            }
        });

        reduceRdd.saveAsTextFile("/output/spark/wcjava");

        sc.stop();
    }
}
